package cn.swanit.iot.verticles;

import cn.hutool.json.JSONUtil;
import io.netty.handler.codec.mqtt.MqttProperties;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Handler;
import io.vertx.core.Promise;
import io.vertx.mqtt.MqttAuth;
import io.vertx.mqtt.MqttServer;
import io.vertx.mqtt.MqttTopicSubscription;
import io.vertx.mqtt.MqttWill;
import io.vertx.mqtt.messages.codes.MqttSubAckReasonCode;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class MqttServerVerticle extends AbstractVerticle {

    private MqttServer mqttServer;

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        this.mqttServer = MqttServer.create(vertx);
        Integer port = config().getInteger("port");
        mqttServer.endpointHandler(endpoint -> {
                    log.info("clientId = [{}] clean session = {}, keep alive timeout = {}", endpoint.clientIdentifier(), endpoint.isCleanSession(), endpoint.keepAliveTimeSeconds());
                    MqttAuth auth = endpoint.auth();
                    if (auth != null) {
                        log.info("MQTT auth {}", auth.toJson());
                    }
                    log.info("connectProperties {}", JSONUtil.toJsonStr(endpoint.connectProperties()));
                    MqttWill mqttWill = endpoint.will();
                    if (mqttWill != null) {
                        log.info("will topic = {}, qos = {}, message = {}, retain = {}", mqttWill.getWillTopic(), mqttWill.getWillQos(), mqttWill.getWillMessage(), mqttWill.isWillRetain());
                    }
                    // accept connection from the remote client
                    endpoint.accept(false);

                    endpoint.disconnectMessageHandler(disconnectMessage -> {
                        log.info("Received disconnect from client, reason code = {}", disconnectMessage.code());
                    });

                    endpoint.subscribeHandler(subscribe -> {
                        log.info("subscribeHandler request {}", subscribe.messageId());
                        List<MqttSubAckReasonCode> reasonCodes = new ArrayList<>();
                        for (MqttTopicSubscription s : subscribe.topicSubscriptions()) {
                            log.info("Subscription for {} with QoS {}", s.topicName(), s.qualityOfService());
                            reasonCodes.add(MqttSubAckReasonCode.qosGranted(s.qualityOfService()));
                        }
                        // ack the subscriptions request
                        endpoint.subscribeAcknowledge(subscribe.messageId(), reasonCodes, MqttProperties.NO_PROPERTIES);
                    }).unsubscribeHandler(unsubscribe -> {
                        log.info("unsubscribeHandler request {}", unsubscribe.messageId());
                        for (String t : unsubscribe.topics()) {
                            log.info("Unsubscription for {}", t);
                        }
                        // ack the subscriptions request
                        endpoint.unsubscribeAcknowledge(unsubscribe.messageId());
                    });

                    endpoint.pingHandler(event -> {
                        log.info("Ping received from client");
                        endpoint.pong();
                    }).closeHandler(event -> log.info("client is closed {}", endpoint.clientIdentifier()));

                    endpoint.publishHandler(message -> {
                        String topicName = message.topicName();
                        log.info("publishHandler {} Just received message [{}] with QoS [{}]", message.messageId(), message.payload().toString(Charset.defaultCharset()), message.qosLevel());
                        if (message.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
                            endpoint.publishAcknowledge(message.messageId());
                        } else if (message.qosLevel() == MqttQoS.EXACTLY_ONCE) {
                            endpoint.publishReceived(message.messageId());
                        }
                        // 发给当前客户端
                        endpoint.publish(topicName, message.payload(), message.qosLevel(), false, false)
                                .onSuccess(packetId -> {
                                    log.info("发布消息成功 {} {}", packetId, topicName);
                                }).onFailure(event -> {
                                    log.error("发布消息失败 {}", event.getMessage());
                                });
                    }).publishReleaseHandler(messageId -> {
                        log.info("发布的消息已接收 {}", messageId);
                        endpoint.publishComplete(messageId);
                    }).publishAcknowledgeHandler(messageId -> {
                        log.info("发布消息确认 {}", messageId);
                    }).publishReceivedHandler(messageId -> {
                        log.info("发布的消息已释放 {}", messageId);
                        endpoint.publishRelease(messageId);
                    }).publishCompletionHandler(messageId -> {
                        log.info("发布完成 {}", messageId);
                    });
                })
                .listen(port)
                .onComplete(successHandler -> {
                    startPromise.complete();
                }, failureHandler -> {
                    startPromise.fail(failureHandler.getCause());
                });

        mqttServer.exceptionHandler(new Handler<Throwable>() {
            @Override
            public void handle(Throwable event) {
                log.error("MqttServerVerticle exceptionHandler", event);
            }
        });
    }

    @Override
    public void stop(Promise<Void> stopPromise) {
        mqttServer.close().onComplete(successHandler -> {
            stopPromise.complete();
        }, failureHandler -> {
            stopPromise.fail(failureHandler.getCause());
        });
    }
}
